# 'Friendly Flow' is a Server Framework for Python 2.5 and up
# Copyright (C) Xaba Software http://www.xaba.nl/
#
# This file is part of 'Friendly Flow'
#
# 'Friendly Flow' is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# 'Friendly Flow' is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with 'Friendly Flow'; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

from unittest import TestCase
from friendlyflow import Scheduler, sleep, read, write, fork

class SchedulerTest(TestCase):
    
    def setUp(self):
        self.selectCalls = []
        self.actualSelectTimeout = "full"
        self.selectReturn = ([], [], [])
        self._clock = 1234
        
    def createScheduler(self, l):
        return Scheduler(l, self.mock_select, self.mock_time)
    
    def testEmpty(self):
        scheduler = self.createScheduler([])
        scheduler.run()
        
    def testOneThread(self):
        l = []
        def thread():
            l.append(0)
            yield
            l.append(1)
            yield
            l.append(2)
        
        scheduler = self.createScheduler([thread()])
        scheduler.run()
        self.assertEquals([0, 1, 2], l)
        
    def testTwoThreads(self):
        l = []
        def thread(name):
            l.append(name + "0")
            yield
            l.append(name + "1")
            yield
            l.append(name + "2")
        
        scheduler = self.createScheduler([thread("a"), thread("b")])
        scheduler.run()
        self.assertEquals(["a0", "b0", "a1", "b1", "a2", "b2"], l)
    
    def testNewSleepingAppend(self):
        #whitebox test
        scheduler = self.createScheduler([])
        scheduler._newSleeping = []
        scheduler._newSleepingAppend(5, "five")
        self.assertEquals([(5, "five")], scheduler._newSleeping)
    
        scheduler._newSleepingAppend(5, "five added later")
        self.assertEquals([(5, "five"), (5, "five added later")], scheduler._newSleeping)

        scheduler._newSleepingAppend(4, "four")
        self.assertEquals([(4, "four"), (5, "five"), (5, "five added later")], scheduler._newSleeping)
    
    def testSleeps(self):
        l = []
        def f(timeout):
            yield sleep(timeout + self.mock_time())
            l.append(timeout)

        before = self.mock_time()
        scheduler = self.createScheduler([f(0.02), f(0.1), f(0.01)])
        scheduler.run()
        after = self.mock_time()
        
        self.assertEquals([0.01, 0.02, 0.1], l)
        self.assertAlmostEqual(0.1, after - before, 5)
        
    def testSleepWithUnblocking(self):
        l = []
        def slow():
            yield sleep(0.01 + self.mock_time())
            l.append("slow")
        def fast():
            yield
            l.append("fast")

        scheduler = self.createScheduler([slow(), fast()])
        scheduler.run()
        self.assertEquals(["fast", "slow"], l)

    def testRead(self):
        descriptor = 23
        l = []
        def f():
            yield read(descriptor)
            l.append("reached after readable")

        scheduler = self.createScheduler([f()])
        
        # yield read(descriptor):
        scheduler.step()
        self.assertEquals([], self.selectCalls)
        
        # tries to read from the yielded read.
        self.selectReturn = ([descriptor], [], [])
        self.actualSelectTimeout = 0.1
        scheduler.step()
        self.assertEquals(([descriptor], [], [], None), self.selectCalls[0])
        self.assertEquals(["reached after readable"], l)

    def testReadWithUnblocking(self):
        descriptor = 12
        l = []
        def f():
            yield read(descriptor)
            l.append("reached after readable")
            
        def g():
            yield
            l.append("after unblocking")

        scheduler = self.createScheduler([f(), g()])
        
        # f: yield read(descriptor); g: yield
        scheduler.step()
        self.assertEquals([], self.selectCalls)
        
        # f tries to read (poll) from the yielded descriptor (not succesful), g should be executed by now
        self.selectReturn = ([], [], []) # the file is not yet readable
        self.actualSelectTimeout = 0.1
        scheduler.step()
        self.assertEquals(([descriptor], [], [], 0), self.selectCalls[0])
        self.assertEquals(["after unblocking"], l)
        
        # f tries to read (blocking) from the yielded descriptor
        self.selectReturn = ([descriptor], [], [])
        scheduler.step()
        self.assertEquals(([descriptor], [], [], None), self.selectCalls[1])
        self.assertEquals(["after unblocking", "reached after readable"], l)

    def testReadWithSleep(self):
        descriptor = 13
        l = []
        def f():
            yield read(descriptor)
            l.append("reached after readable")
            
        def g():
            yield sleep(0.05 + self.mock_time())
            l.append("g0")
            yield sleep(0.05 + self.mock_time())
            l.append("g1")

        scheduler = self.createScheduler([f(), g()])
        
        # f: yield read(descriptor); g: yield
        scheduler.step()
        self.assertEquals([], self.selectCalls)
        
        # f tries to read (poll) from the yielded descriptor (not succesful), g should be executed by now
        self.selectBlockingTimeout = 0.05 # select waits...
        self.selectReturn = ([], [], []) # ...and returns without result
        scheduler.step()
        self.assertEquals(([descriptor], [], []), self.selectCalls[0][:-1])
        self.assertAlmostEqual(0.05, self.selectCalls[0][-1], 3)
        self.assertEquals(["g0"], l)
        
        # f tries to read (blocking) from the yielded descriptor
        self.actualSelectTimeout = 0.02 # select waits shortly...
        self.selectReturn = ([descriptor], [], []) # ...and returns
        scheduler.step()
        self.assertEquals(([descriptor], [], []), self.selectCalls[1][:-1])
        self.assertAlmostEqual(0.05, self.selectCalls[1][-1], 3)
        self.assertEquals(["g0", "reached after readable"], l)

    def testWrite(self):
        descriptor = 11
        l = []
        def f():
            yield write(descriptor)
            l.append("reached after writable")

        scheduler = self.createScheduler([f()])
        
        # yield write(descriptor):
        scheduler.step()
        self.assertEquals([], self.selectCalls)
        
        # tries to write to the yielded write.
        self.selectReturn = ([], [descriptor], [])
        self.actualSelectTimeout = 0.1
        scheduler.step()
        self.assertEquals(([], [descriptor], [], None), self.selectCalls[0])
        self.assertEquals(["reached after writable"], l)
        
    def testFork(self):
        l = []
        def forked():
            l.append("forked 0")
            yield
            l.append("forked 1")
        
        def thread():
            l.append("thread 0")
            yield fork(forked())
            l.append("thread 1")
            yield
            l.append("thread 2")
        
        scheduler = self.createScheduler([thread()])
        scheduler.run()
        self.assertEquals(["thread 0", "forked 0", "thread 1", "forked 1", "thread 2"], l)
        
    def testAllWithoutMocking(self):
        pass

    def mock_select(self, rlist, wlist, xlist, timeout=None):
        if self.actualSelectTimeout == "full":
            self._clock += timeout
        else:
            self._clock += self.actualSelectTimeout
        self.selectCalls.append((rlist, wlist, xlist, timeout))
        return self.selectReturn
    
    def mock_time(self):
        return self._clock


#todo: test multiple readers.

#todo: meaning of multiple readers/writers on the same fileno - is this legal?

#tests with real select and time

#todo implement time nicely

#idea: make an "isblocked" method that uses select to determine if a thread is blocked (for testing purposes)

#another idea: no blocking steps? is this a good idea even just as an option?

#new understanding of read/write: may be done unblocking, resulting in an EAGAIN, only then followed by the select

#todo assertStopIteration
